package com.bff.gaia.client.combine;

import com.bff.gaia.api.common.Plan;
import com.bff.gaia.api.common.operators.GenericDataSinkBase;
import com.bff.gaia.client.cli.ServerHandlers;
import com.bff.gaia.client.program.JobWithJars;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

public class MergePlanClass {

	//存放最初的作业Plan集合
	private List<Plan> plans = new LinkedList<>();

	//上面的 plans 转换为 qPlans
	private List<Qplan> qPlans = new LinkedList<>();

	private List<Qplan> qPlanList=new ArrayList<>();

	//qPlans中能合并的部分放入该链表中，参与合并的任务
	private List<Qplan> mergeInit = new LinkedList<>();

	public List<JobWithJars>  jarlists=new ArrayList<>();

	private  int []visit;

	public List<Qplan> getMergeInit() {
		return mergeInit;
	}

	public void setMergeInit(List<Qplan> mergeInit) {
		this.mergeInit = mergeInit;
	}

	// qPlans中不能合并的部分放入该链表中，不能参与合并的任务
	private List<Plan> preSub = new LinkedList<Plan>();
	private List<Integer> flagList=new ArrayList<>();

	// 合并后的任务放入该链表中
	public  List<Plan> mergePlan = new LinkedList<Plan>();

	// 初始化链表 和 相关元素
	public MergePlanClass(List<Plan> plans, List<Plan> queueP) {
		this.plans = plans;
		System.out.println("Plans的集合大小为："+plans.size());
		for (Plan p : plans) {
			Qplan qp=new Qplan(p);
			if(qp!=null){
				System.out.println("Qplan不为空");
				qPlans.add(qp);
				qPlanList.add(qp);
			}
		}
		int []visited=new int[qPlans.size()];
		for(int k=0; k<visited.length; k++){
			visited[k]=0;
		}
		this.visit=visited;
		System.out.println("QPlan封装完毕=======================，大小为："+qPlans.size());
		System.out.println("visit访问数组的大小为："+this.visit.length);
		// 划分合并型作业和非合并型作业
		for (Qplan qplan : qPlans) {
			if (qplan.isType()) {
				System.out.println("向mergeInit队列中加入作业");
				mergeInit.add(qplan);
			} else {
				System.out.println("向不能合并的作业集合中加入作业");
				preSub.add(qplan.getPlan());
			}
		}
		// 执行并行度合并算法
		// 先根据并行度，从所有可合并作业类型中合并并行度相同的作业
		mergeInit = mergePlanByParallelism();
		System.out.println("mergeInit after merge Plan size is:"+mergeInit.size());
		System.out.println("MergedPlan的大小为："+mergePlan.size());
		/**
		 * tests MergeInit队列中的元素
		 */
		for(int k=0; k<mergeInit.size(); k++){
			/**
			 * PlanName必须是MergePlan类型的
			 */
			System.out.println(mergeInit.get(k).getPlan().getJobName());
		}
		//按文件和槽的大小划分作业类型
		// 合并图结构差异较大的作业
		structureDiffMergeMethod(mergeInit);
		for (int i = 0; i < mergeInit.size(); i++) {
			preSub.add(mergeInit.get(i).getPlan());
		}
		queueP=mergePlan;
		System.out.println("queueP的大小为："+queueP.size());
		System.out.println("输出queueP的作业名称，观察是否是合并并行的作业");
		for(int i=0; i<queueP.size(); i++){
			System.out.println(queueP.get(i).getJobName());
		}
		clearArrayList();
	}

	private void clearArrayList() {
		for(int i = 0; i< ServerHandlers.planList.size(); i++){
			if(!ServerHandlers.visited[i]){
				ServerHandlers.todoPlan.add(ServerHandlers.planList.get(i));
			}
		}
		ServerHandlers.planList.clear();
		ServerHandlers.planList.addAll(ServerHandlers.todoPlan);
		ServerHandlers.todoPlan.clear();
	}

	//合并并行度相同的作业
	private List<Qplan> mergePlanByParallelism() {
		Qplan plan1 = null;
		Qplan plan2 = null;

		List<Qplan> mediaList = new LinkedList<>();

		if (mergeInit.isEmpty()) {
			return mergeInit;
		} else {
			//从mergeInit选出所有并行度相同的作业，合并那些并行度相同的作业，并把剩余的并行度不相同的作业返回
			while (mergeInit.size() > 0) {

				plan1 = mergeInit.get(0);
				mergeInit.remove(0);

				List<Qplan> sameParaWork = new ArrayList<>();
				sameParaWork.add(plan1);

				for (int i = 0; i < mergeInit.size(); i++) {
					plan2 = mergeInit.get(i);
					if (plan1.getParallizer() == plan2.getParallizer()) {
						sameParaWork.add(plan2);
					}
				}
				if (sameParaWork.size() < 2) {
					mediaList.addAll(sameParaWork);
				} else {
					// 合并并行度相同的作业，并把没合并的作业返回
					sameParaWork = paralizerMergeMethod(sameParaWork);
					if (!sameParaWork.isEmpty()) {
						mediaList.addAll(sameParaWork);
					}
				}
			}
		}
		return mediaList;
	}

	//并行度相同合并规则和算法
	private List<Qplan> paralizerMergeMethod(List<Qplan> sameWork) {
		Qplan p1 = null;
		Qplan p2 = null;
		float[][] workRatio = new float[sameWork.size()][3];
		if (sameWork.size() >= 2) {
			p1 = sameWork.get(0);
			//计算 一个任务和其他任务之间的 数值比 启发式
			for (int i = 1; i < sameWork.size(); i++) {
				p2 = sameWork.get(i);
				float fileRatio = ((float) p2.getFileSize()) / ((float) p1.getFileSize());
				if (fileRatio < 1) {
					fileRatio = 1f / fileRatio;
				}
				float deptRatio = ((float) p2.getDept()) / ((float) p1.getDept());
				if (deptRatio < 1) {
					deptRatio = 1f / deptRatio;
				}
				float globalOperatorRatio = ((float) p2.getNumberOfGlobleOperator()) / ((float) p1.getNumberOfGlobleOperator());
				if (globalOperatorRatio < 1) {
					globalOperatorRatio = 1f / globalOperatorRatio;
				}
				workRatio[i][0] = fileRatio;
				workRatio[i][1] = deptRatio;
				workRatio[i][2] = globalOperatorRatio;
			}
			// 从所有并行度相同的任务中挑选出最适合合并的任务
			float minFileRatio = workRatio[1][0];
			float maxDeptRatio = workRatio[1][1];
			int nextMergeWork = 1;
			for (int i = 2; i < sameWork.size(); i++) {
				if (workRatio[i][0] < minFileRatio) {
					if (minFileRatio > 2) {
						minFileRatio = workRatio[i][0];
						maxDeptRatio = workRatio[i][1];
						nextMergeWork = i;
					} else {
						if (workRatio[i][1] > maxDeptRatio) {
							minFileRatio = workRatio[i][0];
							maxDeptRatio = workRatio[i][1];
							nextMergeWork = i;
						}
					}
				}
			}
			p2 = sameWork.get(nextMergeWork);

			sameWork.remove(nextMergeWork);
			sameWork.remove(0);

			mergeTwoPlanSameParall(p1, p2);

			return paralizerMergeMethod(sameWork);

		} else {
			return sameWork;
		}
	}


	//合并两个并行度一样的作业
	private void mergeTwoPlanSameParall(Qplan plan1, Qplan plan2) {
		boolean res=testIfVisited(plan1,plan2);
		if(res){
			return;
		}
		List<GenericDataSinkBase<?>> planSinks = new ArrayList<>();
		for (GenericDataSinkBase<?> out : plan1.getPlan().getDataSinks()) {
			planSinks.add(out);
		}
		for (GenericDataSinkBase<?> out : plan2.getPlan().getDataSinks()) {
			planSinks.add(out);
		}
		Plan plan = new Plan(planSinks);

		plan.setExecutionConfig(plan1.getPlan().getExecutionConfig());

		deleteFromQueuePlan(plan1, plan2);

		combineJarFiles(plan1.getPai().getPlan(),plan2.getPai().getPlan());

		plan.setJobName("MergePlan : " + plan1.getPai().getPlan().getJobName() + " && " + plan2.getPai().getPlan().getJobName());
		System.out.println("Plan合并之后为："+plan.getJobName());
		mergePlan.add(plan);
		findAndVisit(plan1,plan2);
	}

	private boolean testIfVisited(Qplan plan1, Qplan plan2) {
		for(int i=0; i<qPlanList.size(); i++){
			if(qPlanList.get(i).equals(plan1) || qPlanList.get(i).equals(plan2)){
				if(visit[i]==1){
					return true;
				}
			}
		}
		return false;
	}

	private void deleteFromQueuePlan(Qplan plan1, Qplan plan2) {
		/**
		 * 从队列中删除已经合并的元素
		 */

	}

	private void combineJarFiles(Plan plan1, Plan plan2) {
		for(int i = 0; i< ServerHandlers.jarList.size(); i++){
			if(ServerHandlers.planList.get(i)==plan1 || ServerHandlers.planList.get(i)==plan2){
				ServerHandlers.visited[i]=true;
				if(!ServerHandlers.todojars.contains(ServerHandlers.jarList.get(i))){
					/**
					 * 不存在这个Jar文件再加入到todo队列中
					 */
					ServerHandlers.todojars.add(ServerHandlers.jarList.get(i));
				}
			}
		}
		this.jarlists= ServerHandlers.todojars;
	}

	// 合并图结构差异性较大的作业
	private List<Qplan> structureDiffMergeMethod(List<Qplan> workList) {
		Qplan p1 = null;
		Qplan p2 = null;
		if (workList.size() >= 2) {

			p1 = workList.get(0);

			//存放利用启发式方法计算的各种数据
			//数组元素包含的是： 1. 文件大小的比率。 2.
			float[][] workRatio = new float[workList.size()][5];
			float[] workDistence = new float[workList.size()];
			int[][] pOperators = p1.getOperators();
			// 利用启发式方法 计算作业之间的五种数据大小的差异
			for (int i = 1; i < workList.size(); i++) {
				p2 = workList.get(i);
				int[][] operators = p2.getOperators();
				int distence = 0;
				for (int j = 0; j < 19; j++) {
					distence = distence + (pOperators[j][0] - operators[j][0]) * (pOperators[j][0] - operators[j][0]);
				}
				//执行图中算子类型差异比例
				workRatio[i][0] = (float) Math.sqrt(distence);
				distence = 0;

				//两个作业文件大小比率
				float fileRatio = ((float) p2.getFileSize()) / ((float) p1.getFileSize());
				if (fileRatio < 1) {
					fileRatio = 1f / fileRatio;
				}
				//两个作业的深度大小比率
				float deptRatio = ((float) p2.getDept()) / ((float) p1.getDept());
				if (deptRatio < 1) {
					deptRatio = 1f / deptRatio;
				}
				//两个作业全局算子个数的比率

				float globalOperatorRatio = ((float) p2.getNumberOfGlobleOperator()) / ((float) p1.getNumberOfGlobleOperator());
				if (globalOperatorRatio < 1) {
					globalOperatorRatio = 1f / globalOperatorRatio;
				}
				//两个作业并行度大小的比率
				float paraRatio = ((float) p2.getParallizer()) / ((float) p1.getParallizer());
				if (paraRatio < 1) {
					paraRatio = 1f / paraRatio;
				}

				workRatio[i][1] = fileRatio;

				workRatio[i][2] = deptRatio;

				workRatio[i][3] = globalOperatorRatio;

				workRatio[i][4] = paraRatio;

				workDistence[i] = workRatio[i][0] * workRatio[i][1] * workRatio[i][2] * workRatio[i][3];
			}

			// 比较作业差异大小合并作业
			float minFileRatio = workRatio[1][0];
			float minParaRatio = workRatio[1][4];
			float maxDis = workDistence[1];
			int nextMergeWork = 1;

			for (int i = 2; i < workList.size(); i++) {

				float nextFileRatio = workRatio[i][0];
				float nextParaRatio = workRatio[i][4];
				float nextDis = workDistence[i];
				if (nextParaRatio < minParaRatio) {
					//后一个小于二 前一个大于二
					if (nextParaRatio <= 2 && minFileRatio >= 2) {
						minFileRatio = nextFileRatio;
						minParaRatio = nextParaRatio;
						maxDis = nextDis;
						nextMergeWork = i;
					}
					//两个都小于 2
					else if (nextParaRatio <= 2 && minParaRatio <= 2) {
						if (nextFileRatio < minFileRatio) {
							minFileRatio = nextFileRatio;
							minParaRatio = nextParaRatio;
							maxDis = nextDis;
							nextMergeWork = i;
						} else if (nextDis > maxDis) {
							minFileRatio = nextFileRatio;
							minParaRatio = nextParaRatio;
							maxDis = nextDis;
							nextMergeWork = i;
						}
						//如果下一个的并行度大小比率乘以其文件大小比率 比 标志的并行度大小比率乘以文件大小比率小
						// 那么则更换赋值
						else if ((nextParaRatio * nextFileRatio) < (minParaRatio * minFileRatio)) {
							minFileRatio = nextFileRatio;
							minParaRatio = nextParaRatio;
							maxDis = nextDis;
							nextMergeWork = i;
						}
					}
					//两个都大于 2
					else {
						minFileRatio = nextFileRatio;
						minParaRatio = nextParaRatio;
						maxDis = nextDis;
						nextMergeWork = i;
					}
				}
			}
			p2 = workList.get(nextMergeWork);
			workList.remove(nextMergeWork);
			workList.remove(0);
			mergeTwoDiffStrPlan(p1, p2);
			return structureDiffMergeMethod(workList);
		} else {
			return workList;
		}
	}

	//合并两个并行度不同的
	private void mergeTwoDiffStrPlan(Qplan plan1, Qplan plan2) {
		boolean res=testIfVisited(plan1,plan2);
		if(res){
			return;
		}

		List<GenericDataSinkBase<?>> planSinks = new ArrayList<>();
		for (GenericDataSinkBase<?> out : plan1.getPlan().getDataSinks()) {
			planSinks.add(out);
		}
		for (GenericDataSinkBase<?> out : plan2.getPlan().getDataSinks()) {
			planSinks.add(out);
		}
		Plan plan = new Plan(planSinks);

		if(plan1.getParallizer() > plan2.getParallizer()){
			plan.setExecutionConfig(plan1.getPlan().getExecutionConfig());
		}else{
			plan.setExecutionConfig(plan2.getPlan().getExecutionConfig());
		}
		combineJarFiles(plan1.getPai().getPlan(),plan2.getPai().getPlan());
		plan.setJobName("MergePlan : " + plan1.getPai().getPlan().getJobName() + " && " + plan2.getPai().getPlan().getJobName());

		mergePlan.add(plan);
		findAndVisit(plan1,plan2);
	}

	private void findAndVisit(Qplan plan1, Qplan plan2) {
		for(int i=0; i<qPlanList.size(); i++){
			if(qPlanList.get(i).equals(plan1) || qPlanList.get(i).equals(plan2)){
				if(visit[i]==0){
					visit[i]=1;
				}
			}
		}
	}

	//返回 合并作业集合
	public List<Plan> getMergePlan() {
		return mergePlan;
	}
	//返回 未合并作业集合
	public List<Plan> getPreSub() {
		return preSub;
	}
}